package vip.xls8z.reactor.jdk9.reactive;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class FlowDemo {
    public static void main(String[] args) throws InterruptedException {
        // 1. 定义发布者，发布的数据类型是Integer
        // 直接使用JDK自带的SubmissionPublisher，它实现了Publisher接口
        SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();

        // 定义订阅者
        Flow.Subscriber<Integer> subscriber = new Flow.Subscriber<>() {
            Flow.Subscription subscription = null;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                this.subscription.request(1);
            }

            @Override
            public void onNext(Integer item) {
                System.out.println("接收到：" + item);
                this.subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println(throwable.getMessage());
                subscription.cancel();
            }

            @Override
            public void onComplete() {
                System.out.println("处理完成");
            }
        };

        // 发布者和订阅者建立关系
        publisher.subscribe(subscriber);

        // 生产数据并发布
        for (int i = 0; i < 10; i++) {
            publisher.submit(i);
        }

        // 数据处理完，关闭publisher
        publisher.close();

        // 主线程延迟停止
        Thread.sleep(1000);
    }
}
